package net.xuele.learn.flink.book.async;

import com.google.common.collect.Lists;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;


import java.util.concurrent.*;
import java.util.function.Consumer;
import java.util.function.Supplier;

/**
 * @Author patrick
 * @Date 2023/7/18 10:39
 * @Description
 */
public class AsyncDatabaseRequest extends RichAsyncFunction<String, Tuple2<String, String>> {

    private DatabaseClient client;

    @Override
    public void open(Configuration parameters) throws Exception {
        client = new DatabaseClient();
    }

    @Override
    public void close() throws Exception {
        client.close();
    }

    @Override
    public void asyncInvoke(String input, ResultFuture<Tuple2<String, String>> resultFuture) throws Exception {
        Future<String> result = client.query(input);
        CompletableFuture.supplyAsync(new Supplier<String>() {
            @Override
            public String get() {
                try {
                    return result.get();
                } catch (InterruptedException | ExecutionException e) {
                    return null;
                }
            }
        }).thenAccept(dbResult -> resultFuture.complete(Lists.newArrayList(Tuple2.of(input, dbResult))));
    }

    /**
     * 模拟的异步客户端
     */
    private static class DatabaseClient {

        public Future<String> query(String key) {
            return new FutureTask<>(() -> "hello");
        }

        public void close() {

        }
    }
}
